package com.atguigu.edu.realtime.app.dwd.user;

import com.atguigu.edu.realtime.app.BaseAppSql;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdUserCartAdd extends BaseAppSql {
    public static void main(String[] args) {
        new DwdUserCartAdd().init(3073,2,"DwdUserCartAdd");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //1.读取ods_db数据
        sourceFromOdsDb(tEnv,"DwdUserCartAdd");

        //2.过滤加购数据
        Table result = tEnv.sqlQuery("select \n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['course_id'] course_id,\n" +
                "    `data`['course_name'] course_name,\n" +
                "    `data`['cart_price'] cart_price,\n" +
                "    `data`['session_id'] session_id,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `ts`*1000 \n" +
                "from ods_db \n" +
                "where `database` = 'edu' and `table` = 'cart_info' \n" +
                "    and `type` = 'insert'");

        //3.输出到kafka dwd_user_cart_add
        tEnv.executeSql("create table dwd_user_cart_add(\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    course_id string,\n" +
                "    course_name string,\n" +
                "    cart_price string,\n" +
                "    session_id string,\n" +
                "    create_time string,\n" +
                "    ts bigint \n" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_USER_CART_ADD));

        result.executeInsert("dwd_user_cart_add");
    }
}
